#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   DTC_test.py
@Time    :   2022/09/07 13:31:28
@Author  :   YangLiang 
@Version :   1.0
@Contact :   yangliang@chengtech.com
@License :   (C)Copyright 2022-2025
@Desc    :   None
'''

# here put the import lib
from cgitb import enable
from email.message import Message
from pytest_assume.plugin import assume
 
import pytest
import binascii
import logging
import time
import requests
from sympy import I
# from sympy import assuming
from commDriver.TS_CAN import *
import allure
import sys
from collections.abc import Iterable

from sqlalchemy import false
sys.dont_write_bytecode = True
import Set_up

Fault_Injection_DTC_List = [([0xAC, 0xE8, 0x76], 'SensorBlindnessDet'),
                            ([0xAC, 0xE9, 0x76], 'SensorBlindnessDet'),
                            ([0xAC, 0xE6, 0x55], 'VariantCodeDet'),
                            ([0xAC, 0xD1, 0x78], 'AlignmentNeverDone'),
                            ([0xAC, 0xD2, 0x78], 'AlignmentNotDone'),
                            ([0xAC, 0xEA, 0x94], 'VerticalMisaligment'),
                            ([0xAC, 0xEB, 0x94], 'HorizontalMisaligment'),
                            ([0xAC, 0xD4, 0x88], 'FactoryDataFail'),
                            ([0xAC, 0xD9, 0x17], 'Rd3_3OV'),
                            ([0xAC, 0xDB, 0x16], 'Rd3_3UV'),
                            # ([0xAC, 0xDE, 0x00], 'RadarModulation'),
                            ([0xAC, 0xE1, 0x52], 'SwTempFail'),
                            ([0xAC, 0xE2, 0x4B], 'MasterRdTemp'),  # PASS与FAIL状态不是0与-1.需要输入具体数值
                            ([0xAC, 0xED, 0x00], 'AntDiagram'),
                            # ([0xAC, 0xD6, 0x40], 'Watchdog'),                #内部电源管理系统故障 new
                            # ([0xAC, 0xD8, 0x04], 'internalHw'),                #内部硬件故障 new
                            # ([0xAC, 0xDD, 0x00], 'radarModCfg'),                #雷达调制配置故障 new
                            # ([0xAC, 0xDF, 0x52], 'swFail'),                #软件故障 new
                            ([0xAC, 0xE0, 0x00], 'basebandFail'),                #雷达前端数据处理故障 new
                            ]




@allure.severity('critical')
class Test_DTC_Phy(object):

    def dtc_time_check(self,Driver,ex_dtc_list,timeout=5):
        '''
        输入故障码和期望状态
        循环请求 19服务,当读到dtc码或超时5s仍然无DTC码时中断
        '''
        global total_time
        flag = True
        time_start = time.time()
        while flag:
            if time.time() - time_start <= timeout :
                uds_rep,size = Driver['UDS'].req_and_res_can([0x19,0x02,0x09],3,enable_print = False)
                act_dtc_list = []
                if len(uds_rep) >= 7:
                    for i in range(int((len(uds_rep)-3)/4)):
                        act_dtc_list = uds_rep[3+i:7+i]
                        i += 4
                        if set(ex_dtc_list).issubset(set(act_dtc_list)):
                            time_end = time.time()
                            total_time = round(time_end - time_start, 3)
                            readstr = binascii.b2a_hex(bytearray(uds_rep[3:]),' ')
                            logging.info('total time cost :{0} s ---,DTC_status<<<{1}>>>\n'.format(str(total_time),readstr))
                            flag = False
            else:
                flag = False
                time_end = time.time()
                total_time = round(time_end - time_start, 3)
                logging.warning('Overtime, No DTC code output!')
                with assume: assert total_time < timeout 
        return total_time


    
    def clearDTC(self, Driver, iStep,checkinitialized=True):
        # 清除DTC故障
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, [0x14,0xFF,0xFF,0xFF], [0x54], 0, iStep, False, 'UDS', 'noClearFault')
        time.sleep(2)
        if checkinitialized:
            iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, [0x19,0x02,0x09], [0x59,0x02,0x09], 0, iStep, False, 'UDS', 'noClearFault')
            iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        return iStep

    def searchDTC(self, Driver,req, DTC_Number,iStep, isTrue, MatureConditionsTime=1000, timeout=0.5):
        # 查询故障
        with allure.step(f"Step{iStep}: search DTC {binascii.b2a_hex(bytearray(DTC_Number), ' ')}"):
            time.sleep(MatureConditionsTime/1000)
            a,b,c=Driver['UDS'].search_DTC(req, DTC_Number, timeout)
            if isTrue:
                if (b + MatureConditionsTime/1000) > (MatureConditionsTime/1000 + 0.25):
                    logging.error(f"MatureConditionsTime = {MatureConditionsTime/1000 + 0.1}, realtime = {b + MatureConditionsTime/1000}")
                if a != True:
                    logging.error(f"No found DTC{binascii.b2a_hex(bytearray(DTC_Number), ' ')}")
                with assume: assert a == True and (b + MatureConditionsTime/1000) <= (MatureConditionsTime/1000 + 0.25)
            else:
                with assume: assert a == False
            iStep +=1
        return iStep

    '''设置信号使能状态'''
    def setSignalEanble(self, Driver, message, status, iStep):
        with allure.step(f'Step{iStep}: set the {message} status to {status}'):
            Driver['CAN'].Enable_rbs_message('CAKY_FRM','BCM', message, status)
        iStep += 1
        return iStep


    '''设置信号值'''
    def setSignalValue(Self, Driver, message, signal, value, iStep):
        with allure.step(f'Step{iStep}: set the value of the {signal} in {message} to {value}'):
            Driver['CAN'].write_db_signal('CAKY_FRM','BCM', message, signal, value)
            time.sleep(0.2)
            iStep += 1
            readValue = Driver['CAN'].read_db_signal('CAKY_FRM','BCM', message, signal)
        return iStep
            

    def checkDTC(self, Driver, DTCNumber, message, cycleTime, missCount, iStep, signal='', faultValue=0, normalValue=0, settype='enable'):
        req = [0x19,0x02,0x09]
        DTC_Number_Exist = DTCNumber + [0x09]
        DTC_Number_NoExist = DTCNumber + [0x08]
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        iStep = self.clearDTC(Driver, iStep)
        time.sleep(1.5)

        if settype == 'enable':
            # 停发报文
            iStep = self.setSignalEanble(Driver,message, False,iStep)
            if message == 'MPC_0x374':
                iStep = self.setSignalEanble(Driver,'MPC_0x295', False,iStep)
                readValue = Driver['CAN'].read_db_signal('CAKY_FRM','BCM', 'Left_BCM_0x12D', 'BCMPower_Gear_12D_S')
                readValue = Driver['CAN'].read_db_signal('CAKY_FRM','BCM', 'ic_odo_info', 'ic_pm_odo')
            if message == 'MPC_0x295':
                iStep = self.setSignalEanble(Driver,'MPC_0x374', False,iStep)
                readValue = Driver['CAN'].read_db_signal('CAKY_FRM','BCM', 'Left_BCM_0x12D', 'BCMPower_Gear_12D_S')
                readValue = Driver['CAN'].read_db_signal('CAKY_FRM','BCM', 'ic_odo_info', 'ic_pm_odo')

        else:
            iStep = self.setSignalValue(Driver, message, signal, faultValue, iStep)
        # 查询故障是否已经产生
        iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, cycleTime * missCount + 100)

        if settype == 'enable':
            # 停发报文
            iStep = self.setSignalEanble(Driver,message, True,iStep)
        else:
            iStep = self.setSignalValue(Driver, message, signal, normalValue, iStep)
        # 查询故障是否已经恢复
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, cycleTime + 100, 1)
        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        # iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, 3000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, False, 3000, 1)

        return iStep


    @allure.title('Test master under voltage DTC in default session')
    @allure.description('Test master under voltage DTC in default session with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_Master_Undervoltage_DTC_Default_Phy(self,action,BCMPower_Gear,Driver):
        req = [0x19,0x02,0x09]
        DTC_Number_Exist = [0xAC,0xCF,0x16,0x09]
        DTC_Number_NoExist = [0xAC,0xCF,0x16,0x08]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        iStep = self.clearDTC(Driver, iStep)
        # 设置电压为8v 制造低压故障
        iStep = Driver['CAN'].SetVolt(Driver, 8, 0.5, iStep)
        # 查询低压故障是否已经产生
        iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, 1000)
        # 恢复电压为13v
        iStep = Driver['CAN'].SetVolt(Driver, 13, 0.5, iStep)
        # 查询故障是否已经恢复
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, 1000)

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        # iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, 3000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, False, 3000, 1)

    @allure.title('Test master the voltage boundary value generate DTC code in default session')
    @allure.description('Test master voltage boundary DTC in default sessionin with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_Master_Undervoltage_boundary_DTC_Phy(self,action,BCMPower_Gear,Driver):

        req = [0x19,0x02,0x09]
        DTC_Number_Exist = [0xAC,0xCF,0x16,0x09]
        DTC_Number_NoExist = [0xAC,0xCF,0x16,0x08]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)

        # 无故障到有故障
        for i in range(6):
            voltage = 8.9 - i/10
            
            iStep = Driver['CAN'].SetVolt(Driver, 12, 0.5, iStep)
            iStep = self.clearDTC(Driver, iStep)
            time.sleep(0.5)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 0.5, iStep)
            if voltage < 8.6:
                iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, 1000)
        
        # 有故障到无故障
        for i in range(6):
            voltage = 8.3 + i/10
            iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
            iStep = self.clearDTC(Driver, iStep)
            time.sleep(1)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 0.5, iStep)
            if voltage < 8.6:
                iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, 1000)
        
        iStep = Driver['CAN'].SetVolt(Driver, 12, 0.5, iStep)
        iStep = self.clearDTC(Driver, iStep)


    @allure.title('Test Master over voltage DTC in default session')
    @allure.description('Test Master over voltage DTC in default sessionin with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_Master_Overvoltage_DTC_Default_Phy(self,action,BCMPower_Gear,Driver):
        req = [0x19,0x02,0x09]
        DTC_Number_Exist = [0xAC,0xCE,0x17,0x09]
        DTC_Number_NoExist = [0xAC,0xCE,0x17,0x08]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        iStep = self.clearDTC(Driver, iStep)
        iStep = Driver['CAN'].SetVolt(Driver, 17, 0.6, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, 1000)

        iStep = Driver['CAN'].SetVolt(Driver, 12, 0.6, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, 1000)
        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)
        # iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, 1000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, False, 1000, 1)

    @allure.title('Test master the voltage boundary value generate DTC code in default session')
    @allure.description('Test master voltage boundary DTC in default sessionin with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_Master_Overvoltage_boundary_DTC_Phy(self,action,BCMPower_Gear,Driver):

        req = [0x19,0x02,0x09]
        DTC_Number_Exist = [0xAC,0xCE,0x17,0x09]
        DTC_Number_NoExist = [0xAC,0xCE,0x17,0x08]

        iStep = 1
        iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '1001', '5001', 0, iStep)
        # 无故障到有故障
        for i in range(6):
            voltage = 16.0 + i/10
            iStep = self.clearDTC(Driver, iStep,False)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 0.5, iStep)
            if voltage > 16.3:
                iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, 1000)
        iStep = Driver['CAN'].SetVolt(Driver, voltage, 1, iStep)
        # 有故障到无故障
        for i in range(5):
            voltage = 16.5 - i/10
            iStep = self.clearDTC(Driver, iStep,False)
            iStep = Driver['CAN'].SetVolt(Driver, voltage, 0.5, iStep)
            if voltage > 16.3:
                iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, 1000)
        
        iStep = Driver['CAN'].SetVolt(Driver, 12, 1, iStep)
        iStep = self.clearDTC(Driver, iStep)




    @allure.title('Test production mode active DTC in default session')
    @allure.description('Test production mode active in default session with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_ProductionModeActive_DTC_Default_Phy(self,action,BCMPower_Gear,Driver):
        req = [0x19,0x02,0x09]
        DTC_Number_Exist = [0xAC,0xD0,0x00,0x09]
        DTC_Number_NoExist = [0xAC,0xD0,0x00,0x08]

        iStep = 1

        iStep = self.clearDTC(Driver, iStep)
        
        # 更改模式为生产模式
        with allure.step(f'Step{iStep}: set production mode is factory mode'):
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '2E340101', '6E3401', 3, iStep, True)
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '223401', '62340101', 0, iStep, False)

        # 查询故障是否已经产生
        iStep = self.searchDTC(Driver, req, DTC_Number_Exist, iStep, True, 1000)
        # 更改模式为正常模式
        with allure.step(f'Step{iStep}: set production mode is normal mode'):
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '2E340100', '6E3401', 3, iStep, True)
            iStep,uds_resp, size = Driver['CAN'].checkReqResp(Driver, '223401', '62340100', 0, iStep, False)
        # 查询故障是否已经恢复
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, 1000)
        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)
        
        # iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, True, 3000, 1)
        iStep = self.clearDTC(Driver, iStep)
        iStep = self.searchDTC(Driver, req, DTC_Number_NoExist, iStep, False, 3000, 1)


    @allure.title('Test Left_BCM_CRC DTC by 0x121 Checksum Error in default session')
    @allure.description('Test Left_BCM_CRC DTC by 0x121 Checksum Error in default sessionin with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_DTC_SFCAN_BUSOFF_Default_Phy(self,action,BCMPower_Gear,Driver):
        iStep = 1
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, [0x19,0x02,0x09], [0x59,0x02,0x09], 1, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)

        Driver['CAN'].V_ConnectCanH_CanL(True, iStep)
        Driver['CAN'].V_ConnectCanH_CanL(False, iStep)
        iStep = self.searchDTC(Driver, [0x19,0x02,0x09], [0xC1,0xD7,0x88,0x08], iStep, True, 300, 1)
        time.sleep(1)
        iStep = self.searchDTC(Driver, [0x19,0x02,0x09], [0xC1,0xD7,0x88,0x08], iStep, True, 100, 1)

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        # iStep = self.searchDTC(Driver, [0x19,0x02,0x09], [0xC1,0xD7,0x88,0x08], iStep, True, 1000, 1)
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, [0x14,0xFF,0xFF,0xFF], [0x54], 1, iStep, False, 'UDS', 'noClearFault')
        iStep = self.searchDTC(Driver, [0x19,0x02,0x09], [0xC1,0xD7,0x88,0x08], iStep, False, 1000, 1)



    ids = ["Request {0}".format(binascii.b2a_hex(bytearray(req), ' '), dtcName ) for req, dtcName in Fault_Injection_DTC_List]
    @allure.title('Test Injection_DTC in default session')
    @allure.description('Test Injection_DTC in default sessionin with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    @pytest.mark.parametrize('req,dtcName', Fault_Injection_DTC_List, ids=ids)
    def test_Injection_DTC_Default_Phy(self, action,BCMPower_Gear, Driver, req, dtcName):
        iStep = 1
        iStep,uds_resp,size = Driver['CAN'].checkReqResp(Driver, [0x19,0x02,0x09], [0x59,0x02,0x09], 1, iStep, False, 'UDS', 'ClearFault')
        iStep = Driver['CAN'].delInitialFailure(iStep,uds_resp)
        iStep = Driver['CAN'].setVariantCodeDet(Driver,dtcName, -1, iStep)
        time.sleep(0.5)
        if req == [0xAC, 0XE8, 0X76]:
            # 要求行驶里程大于5公里
            Driver['CAN'].write_db_signal('CAKY_FRM', 'BCM', 'ic_odo_info', 'ic_pm_odo',10)
            Driver['CAN'].write_db_signal('CAKY_FRM', 'BCM', 'abs_sts_run1', 'abs_pm_veh_spd', 20)
            Driver['CAN'].read_db_signal('CAKY_FRM', 'BCM', 'abs_sts_run1', 'abs_pm_veh_spd')
            time.sleep(1)
            Driver['CAN'].write_db_signal('CAKY_FRM', 'BCM', 'ic_odo_info', 'ic_pm_odo',16)

        a,b,c=Driver['UDS'].search_DTC([0x19,0x02,0x09], req + [0x09])
        if a != True:
            logging.error(f"No found DTC{binascii.b2a_hex(bytearray(req + [0x09]), ' ')}")
        with assume: assert a == True

        iStep = Driver['CAN'].setVariantCodeDet(Driver,dtcName, 0,iStep)
        if req == [0xAC, 0XE8, 0X76]:
            Driver['CAN'].write_db_signal('CAKY_FRM', 'BCM', 'ic_odo_info', 'ic_pm_odo',0)
            Driver['CAN'].write_db_signal('CAKY_FRM', 'BCM', 'abs_sts_run1', 'abs_pm_veh_spd', 0)
            Driver['CAN'].read_db_signal('CAKY_FRM', 'BCM', 'abs_sts_run1', 'abs_pm_veh_spd')
        time.sleep(0.5)
        a,b,c=Driver['UDS'].search_DTC([0x19,0x02,0x09], req + [0x08])
        if a != True:
            logging.error(f"DTC [{binascii.b2a_hex(bytearray(req + [0x08]), ' ')}] not recovered")
        with assume: assert a == True

        # 重启产品
        iStep = Driver['CAN'].restartMRR_Power(Driver,iStep)

        # if req == [0xAC, 0xE6, 0x55] or req == [0xAC, 0xD1, 0x78] or req == [0xAC, 0xD2, 0x78]:
        #     a,b,c=Driver['UDS'].search_DTC([0x19,0x02,0x09], req + [0x08],timeout=1)
        #     if a == False:
        #         a,b,c=Driver['UDS'].search_DTC([0x19,0x02,0x09], req + [0x09],timeout=1)
        # else:
        #     a,b,c=Driver['UDS'].search_DTC([0x19,0x02,0x09], req + [0x08],timeout=1)
        # if a != True:
        #     logging.error(f"Restart Device DTC [{binascii.b2a_hex(bytearray(req), ' ')}] is not saved")
        # with assume: assert a == True and b <= 1

        uds_rep,size = Driver['UDS'].req_and_res_can([0x14,0xFF,0xFF,0xFF],4)
        with assume: assert uds_rep[0] == 0x54
        time.sleep(1.5)
        a,b,c=Driver['UDS'].search_DTC([0x19,0x02,0x09], req + [0x08],timeout=1)
        with assume: assert a == False

    @allure.title('Test DTC MPC TIMEOUT DTC by lost 0x11F message in default session')
    @allure.description('Test DTC MPC timeout DTC by lost 0x11F message in default session with physical addressing')
    @allure.testcase('Test cases address:https://chengtech.pingcode.com/testhub/libraries/630357052177ea2394680877/cases')
    def test_DTCMPCTimeOut_DTC_Lost_0x141_Default_Phy(self, action,BCMPower_Gear, Driver):
        DTCNumber = [0xD1, 0x18, 0x86]
        iStep = 1
        iStep = self.checkDTC(Driver, DTCNumber, 'esc_sts_run_1', 20, 10, iStep)
        